/*
 * Copyright (C) 2008 Universidade Federal de Campina Grande
 *  
 * This file is part of OurGrid. 
 *
 * OurGrid is free software: you can redistribute it and/or modify it under the
 * terms of the GNU Lesser General Public License as published by the Free 
 * Software Foundation, either version 3 of the License, or (at your option) 
 * any later version. 
 * 
 * This program is distributed in the hope that it will be useful, but WITHOUT 
 * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or 
 * FITNESS FOR A PARTICULAR PURPOSE. See the GNU Lesser General Public License
 * for more details. 
 * 
 * You should have received a copy of the GNU Lesser General Public License 
 * along with this program. If not, see <http://www.gnu.org/licenses/>.
 * 
 */
package org.ourgrid.peer.controller;

import static org.ourgrid.common.util.CommonUtils.checkKey;

import java.io.IOException;
import java.util.LinkedList;
import java.util.List;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;

import org.ourgrid.common.LoginResult;
import org.ourgrid.common.interfaces.AccountingAggregator;
import org.ourgrid.common.interfaces.LocalWorkerProvider;
import org.ourgrid.common.interfaces.LocalWorkerProviderClient;
import org.ourgrid.common.interfaces.to.LocalWorkerState;
import org.ourgrid.common.interfaces.to.RequestSpec;
import org.ourgrid.common.statistics.control.JobControl;
import org.ourgrid.common.statistics.control.LoginControl;
import org.ourgrid.common.statistics.control.UserControl;
import org.ourgrid.common.statistics.control.WorkerControl;
import org.ourgrid.peer.PeerConfiguration;
import org.ourgrid.peer.PeerConstants;
import org.ourgrid.peer.controller.accounting.AccountingCommitController;
import org.ourgrid.peer.controller.allocation.DefaultAllocator;
import org.ourgrid.peer.controller.allocation.RedistributionController;
import org.ourgrid.peer.controller.messages.RequestMessages;
import org.ourgrid.peer.controller.messages.WorkerMessages;
import org.ourgrid.peer.dao.AllocationDAO;
import org.ourgrid.peer.dao.ConsumerDAO;
import org.ourgrid.peer.dao.LocalWorkersDAO;
import org.ourgrid.peer.dao.RequestDAO;
import org.ourgrid.peer.dao.UsersDAO;
import org.ourgrid.peer.to.AllocableWorker;
import org.ourgrid.peer.to.LocalConsumer;
import org.ourgrid.peer.to.LocalWorker;
import org.ourgrid.peer.to.PeerUser;
import org.ourgrid.peer.to.PeerUserReference;
import org.ourgrid.peer.to.Request;
import org.ourgrid.reqtrace.Req;

import br.edu.ufcg.lsd.commune.api.InvokeOnDeploy;
import br.edu.ufcg.lsd.commune.api.MonitoredBy;
import br.edu.ufcg.lsd.commune.container.servicemanager.ServiceManager;
import br.edu.ufcg.lsd.commune.identification.ContainerID;
import br.edu.ufcg.lsd.commune.identification.DeploymentID;
import br.edu.ufcg.lsd.commune.identification.ServiceID;

/**
 * Performs Local Worker Provider Receiver actions
 */
public class LocalWorkerProviderController implements LocalWorkerProvider {

	private ServiceManager serviceManager;

	@InvokeOnDeploy
	public void init(ServiceManager serviceManager) {
		this.serviceManager = serviceManager;
	}
	
	
	/**
	 * Processes the login request of a MyGrid client.
	 * Returns to the client the result of the login operation, and in case of successful login
	 * the stub to the Accounting object.
	 * @param workerProviderClient The client callback
	 * @param password The password of the client requesting login
	 * @param myGridPublicKey The PublicKey Object of the MyGrid requesting login
	 */
	@Req("REQ108")
	public void login(
			@MonitoredBy(PeerConstants.CLIENT_MONITOR_OBJECT_NAME)
			LocalWorkerProviderClient workerProviderClient, String password) {
		
		String brokerPublicKey = serviceManager.getSenderPublicKey();
		
		LocalWorkerProvider localWorkerProvider = (LocalWorkerProvider) serviceManager.getObjectDeployment(
				PeerConstants.LOCAL_ACCESS_OBJECT_NAME).getObject();
		
		DeploymentID workerProviderClientID = serviceManager.getStubDeploymentID(workerProviderClient);
		
		ContainerID accessPointID = workerProviderClientID.getContainerID();
		String login = accessPointID.getUserName()+"@"+accessPointID.getServerName();
		
		PeerUser user = UserControl.getInstance().getUser(login, getUsersDAO());

		AccountingAggregator accounting = (AccountingAggregator) serviceManager.getObjectDeployment(
				PeerConstants.ACCOUNTING_OBJECT_NAME).getObject();

		if(user == null){
			LoginResult loginResult = new LoginResult(LoginResult.UNKNOWN_USER);
			workerProviderClient.loginSucceed(localWorkerProvider, loginResult, accounting);
			user = new PeerUser(accessPointID.getUserName(), "", accessPointID.getServerName(), 
					"", false);
			
			getLoginStatistics().login(user, loginResult, serviceManager);
			serviceManager.release(workerProviderClient);
			return;
		}
		
		if(!checkPassword(user,password)){
			LoginResult loginResult = new LoginResult(LoginResult.WRONG_PASSWORD);
			workerProviderClient.loginSucceed(localWorkerProvider, loginResult, accounting);
			getLoginStatistics().login(user, loginResult, serviceManager);
			serviceManager.release(workerProviderClient);
			return;
		}
		
		if(isFirstLogin(user)){
				try {
					UserControl.getInstance().registerPublicKey(user, brokerPublicKey, getUsersDAO());
					user.setPublicKey(brokerPublicKey);
				} catch (IOException e) {
					serviceManager.getLog().error(e);
					LoginResult loginResult = new LoginResult(LoginResult.INTERNAL_ERROR);
					workerProviderClient.loginSucceed(localWorkerProvider, loginResult, accounting);
					getLoginStatistics().login(user, loginResult, serviceManager);
					serviceManager.release(workerProviderClient);
				}
		}
		
		if(!checkKey(user.getPublicKey(), brokerPublicKey)){
			LoginResult loginResult = new LoginResult(LoginResult.WRONG_PUBLIC_KEY);
			workerProviderClient.loginSucceed(localWorkerProvider, loginResult, accounting);
			getLoginStatistics().login(user, loginResult, serviceManager);
			serviceManager.release(workerProviderClient);
			return;
		}
		
		PeerUserReference loggedUser = getUsersDAO().getLoggedUser(user.getPublicKey());
		
		if(loggedUser != null && !loggedUser.getDeploymentId().equals(workerProviderClientID)){
			LoginControl.getInstance().localConsumerFailure(user, serviceManager);
		}
		
		WorkerProviderClientFailureController.finishUserRequests(serviceManager, brokerPublicKey);

		if(loggedUser != null){
			DeploymentID loggedUserID = loggedUser.getDeploymentId();
			if (loggedUserID.equals(workerProviderClientID)) {
				LoginResult loginResult = new LoginResult(); //TODO LoginResult.ALREADY_LOGGED);
				workerProviderClient.loginSucceed(localWorkerProvider, loginResult, accounting);
				getLoginStatistics().login(user, loginResult, serviceManager);
				return;
			}
			
			getUsersDAO().removeLoggedUser(brokerPublicKey);
			serviceManager.getDAO(ConsumerDAO.class).removeLocalConsumer(brokerPublicKey);
			
		}
		
		doLogin(workerProviderClient, localWorkerProvider, brokerPublicKey, user, workerProviderClientID);
	}

	/**
	 * @return
	 */
	private LoginControl getLoginStatistics() {
		return LoginControl.getInstance();
	}


	@Req("REQ108")
	private void doLogin(LocalWorkerProviderClient workerProviderClient, LocalWorkerProvider localWorkerProvider, 
			String publicKey, PeerUser user, DeploymentID workerProviderClientID) {
		
		getUsersDAO().addLoggedUser(publicKey, new PeerUserReference(workerProviderClient, 
				workerProviderClientID));
		
		AccountingAggregator accounting = (AccountingAggregator) serviceManager.getObjectDeployment(
				PeerConstants.ACCOUNTING_OBJECT_NAME).getObject();
		
		LoginResult loginResult = new LoginResult();
		workerProviderClient.loginSucceed(localWorkerProvider, loginResult, accounting);
		
		getLoginStatistics().login(user, loginResult, serviceManager);
	}

	@Req("REQ108")
	private boolean isFirstLogin(PeerUser user) {
		return user.getPublicKey().equals("");
	}

	@Req("REQ108")
	private boolean checkPassword(PeerUser user, String password) {
		return user.getPassword().equals(password);
	}

	@Req("REQ011")
	public void requestWorkers(RequestSpec requestSpec) {
		requestWorkers(requestSpec, serviceManager.getSenderPublicKey());
	}
	
	/**
	 * Validates and executes a local request.
	 * 
	 * @param requestSpec
	 * @param userPubKey
	 */
	@Req("REQ011")
	public void requestWorkers(RequestSpec requestSpec, String userPubKey) {
		RequestDAO reqDao = getRequestDAO();
		//New request from MyGrid
		if(UserControl.getInstance().userExists(userPubKey, getUsersDAO())) {
			
			if (!validateRequest(requestSpec, userPubKey)) {
				return;
			}
			
			PeerUserReference loggedUser = getUsersDAO().getLoggedUser(userPubKey);
			
			if (loggedUser == null) {
				serviceManager.getLog().warn(RequestMessages.getUserDownMessage(requestSpec.getRequestId(), 
						userPubKey));
				return;
			}
			
			LocalWorkerProviderClient localWorkerProviderClient = loggedUser.getClient();
			DeploymentID workerProviderClientID = serviceManager.getStubDeploymentID(localWorkerProviderClient);
			
			if (workerProviderClientID == null) {
				serviceManager.getLog().warn(RequestMessages.getUserDownMessage(requestSpec.getRequestId(), 
						userPubKey));
				return;
			}
			
			//Do not allow Mygrid to repeat the request
			if(requestIsRunning(requestSpec)) {
				serviceManager.getLog().warn(RequestMessages.getRequestIDAlreadyExistsMessage(requestSpec.getRequestId(), 
						workerProviderClientID));
				return;
			} 
			
			serviceManager.getLog().info(RequestMessages.getNewRequestMessage(requestSpec, workerProviderClientID));
			
	        PeerUser user = UserControl.getInstance().getUserByPublicKey(workerProviderClientID.getPublicKey(), getUsersDAO());
			LocalConsumer localConsumer = getConsumerDAO().createLocalConsumer(user);

			Request request = reqDao.createRequest(workerProviderClientID, localWorkerProviderClient, 
					requestSpec, localConsumer);

			doRequest(requestSpec, localWorkerProviderClient, userPubKey, request, false);

		//Request repetition
		} else if (serviceManager.isThisMyPublicKey(userPubKey)) {
			
			Request request = reqDao.getRequest(requestSpec.getRequestId());
			
			if(request != null) {
				DeploymentID consumerID = request.getConsumer().getDeploymentID();
				LocalWorkerProviderClient localWorkerProviderClient = getUsersDAO().getLoggedUser(consumerID.getPublicKey()).getClient();
				doRequest(requestSpec, localWorkerProviderClient, consumerID.getPublicKey(), request, true);
			}
			
		} else {
			serviceManager.getLog().warn(RequestMessages.getUnknownUserMessage(requestSpec, userPubKey));
		}
		
	}

	/**
	 * Executes a request. This method does not validate the request. This
	 * code handles the request and request repetition actions. 
	 * 
	 * @param requestSpec
	 * @param localWorkerProviderClient
	 * @param userPublicKey 
	 * @param request 
	 */
	public void doRequest(RequestSpec requestSpec, LocalWorkerProviderClient localWorkerProviderClient, 
			String userPublicKey, Request request, boolean isRepetition) {
		
		DefaultAllocator allocator = DefaultAllocator.getInstance();

		List<AllocableWorker> allAllocableWorkers = serviceManager.getDAO(AllocationDAO.class).getAllAllocableWorkers();
		List<AllocableWorker> myAllocableWorkersSorted = allocator.getAllocableWorkersForLocalRequest(
				request, allAllocableWorkers, serviceManager);

        LocalConsumer localConsumer = getConsumerDAO().getLocalConsumer(userPublicKey);

		for (AllocableWorker allocableWorker : myAllocableWorkersSorted) {
            dispatchAllocation(localWorkerProviderClient, request, allocableWorker, localConsumer);
		}
		
		if( request.getNeededWorkers() > 0 ) {
			forwardToCommunity(requestSpec);
			
			if (!serviceManager.isThisMyPublicKey(userPublicKey)) {
				scheduleRequest(requestSpec);
			}
		}
		
		if (!isRepetition) {
			JobControl.getInstance().addRequest(request, serviceManager);
		}
	}

	private boolean requestIsRunning(RequestSpec requestSpec) {
		return getRequestDAO().isRunning(requestSpec);
	}

	/**
	 * @param localWorkerProviderClient
	 * @param request
	 * @param allocableWorker
	 * @param userPublicKey 
	 */
	private void dispatchAllocation(LocalWorkerProviderClient localWorkerProviderClient, Request request, 
										AllocableWorker allocableWorker, LocalConsumer localConsumer) {
		
		if (allocableWorker.getConsumer() != null) {//preemption
			Object workerManagement = allocableWorker.getWorkerManagement();
			
			DeploymentID workerManagementID = serviceManager.getStubDeploymentID(workerManagement);
			
			serviceManager.getLog().info("Request "+request.getSpec().getRequestId()+": Taking worker ["
					+ workerManagementID + "] from ["+
					allocableWorker.getConsumer().getDeploymentID()+"]");
			
			if (allocableWorker.getConsumer().isLocal()) {
				Request loserRequest = allocableWorker.getRequest();
				loserRequest.removeAllocableWorker(allocableWorker);
				
				RequestSpec spec = loserRequest.getSpec();
				if (!getRequestDAO().containsAScheduledRequest(spec) && loserRequest.needMoreWorkers() ) {
					scheduleRequest(spec);
				}
			}
			
			serviceManager.registerInterest(allocableWorker.getMonitorName(), 
					workerManagementID.getServiceID().toString(), allocableWorker.getMonitorableType());
		}
		
		allocableWorker.clear();
		
		allocableWorker.setStatus(LocalWorkerState.IN_USE);
		
		if (allocableWorker.isWorkerLocal()) {
			WorkerControl.getInstance().statusChanged(
					allocableWorker.getWorkerSpec().getServiceID(), 
					LocalWorkerState.IN_USE, 
					localConsumer.getDeploymentID().getContainerID().getUserAtServer(), 
					serviceManager);
		}	
		
		allocableWorker.setConsumer(localConsumer);
		allocableWorker.setRequest(request);
        request.addAllocableWorker(allocableWorker);
		
        DeploymentID lwpcDID = serviceManager.getStubDeploymentID(localWorkerProviderClient);
        
		allocableWorker.workForMyGrid(lwpcDID);
	}

	/**
	 * @param requestSpec
	 */
	private void scheduleRequest(RequestSpec requestSpec) {
		
		RequestDAO requestDAO = getRequestDAO();
		
		if(! requestDAO.containsAScheduledRequest(requestSpec)) {
			int delay = serviceManager.getContainerContext().parseIntegerProperty(PeerConfiguration.PROP_REPEAT_REQUEST_DELAY);
			
			long requestID = requestSpec.getRequestId();
			Future<?> future = serviceManager.scheduleActionWithFixedDelay(
					PeerConstants.REQUEST_WORKERS_ACTION_NAME, 
					delay, TimeUnit.SECONDS, requestID);
			
			requestDAO.putScheduledRequest(requestSpec, future);
			serviceManager.getLog().debug(RequestMessages.getRequestRepetitionMessage(delay, requestID));
		}
	}

	/**
	 * @param requestSpec
	 */
	private void forwardToCommunity(RequestSpec requestSpec) {
		CommunityObtainerController.getInstance().request(requestSpec, serviceManager);
	}

	/**
	 * @param requestSpec
	 * @param userPubKey
	 */
	@Req("REQ011")
	private boolean validateRequest(RequestSpec requestSpec, String userPubKey) {
		PeerUserReference loggedUser = getUsersDAO().getLoggedUser(userPubKey);
		
		if (loggedUser == null) {
			return false;
		}
		
		DeploymentID userID = loggedUser.getDeploymentId();

		if(requestSpec == null) {
			serviceManager.getLog().warn(RequestMessages.getNullRequestMessage(userID));
			return false;
		}
		
		if(requestSpec.getRequiredWorkers() < 1) {
			serviceManager.getLog().warn(RequestMessages.getRequestWithLessThanOneWorkerMessage(requestSpec, userID));
			return false;
		}
		
		return true;
	}

	@Req("REQ116")
	public void updateRequest(RequestSpec requestSpec) {
		
		String userPubKey = serviceManager.getSenderPublicKey();
		
		if (!UserControl.getInstance().userExists(userPubKey, getUsersDAO())) {
			serviceManager.getLog().warn(RequestMessages.getUnknownConsumerUpdatingRequestMessage(userPubKey));
			return;
		}
		
		PeerUserReference loggedUser = getUsersDAO().getLoggedUser(userPubKey);
		if (loggedUser == null) {
			serviceManager.getLog().warn(RequestMessages.getUnknownConsumerUpdatingRequestMessage(userPubKey));
			return;
		}
		
		LocalWorkerProviderClient localWorkerProviderClient = loggedUser.getClient();
		
		DeploymentID lwpcObjectID = serviceManager.getStubDeploymentID(localWorkerProviderClient);
		
		if(requestSpec == null) {
			serviceManager.getLog().warn(RequestMessages.getNullRequestUpdateMessage(lwpcObjectID));
			return;
		}
		
		long requestID = requestSpec.getRequestId();
		Request request = getRequestDAO().getRequest(requestID);
		
		if (isRequestUnknown( userPubKey, request )) {			
			serviceManager.getLog().warn(RequestMessages.getUnknownRequestUpdateMessage(lwpcObjectID, requestID));
			return;
		}
		
		if(requestSpec.getRequiredWorkers() < 1) {
			serviceManager.getLog().warn(RequestMessages.getNonPositiveNoOfWorkersMessage(lwpcObjectID, requestID));
			return;
		}
		
		Future<?> future = getRequestDAO().removeFuture(requestSpec);
		
		if (future != null) {
			future.cancel(true);
		}

		request.setSpec(requestSpec);
		
		if (request.needMoreWorkers()) {
			scheduleRequest(requestSpec);
		}
	}


	private boolean isRequestUnknown( String userPubKey, Request request ) {
		return request == null || !request.getConsumer().getPublicKey().equals(userPubKey);
	}

	@Req("REQ117")
	public void pauseRequest(long requestID) {
		
		String senderPubKey = serviceManager.getSenderPublicKey();
		
		if (!UserControl.getInstance().userExists(senderPubKey, getUsersDAO())) {
			serviceManager.getLog().warn(RequestMessages.getUnknownConsumerPausingRequestMessage(senderPubKey));
			return;
		}

		Request request = getRequestDAO().getRequest(requestID);
		PeerUserReference loggedUser = getUsersDAO().getLoggedUser(senderPubKey);
		
		if (loggedUser == null) {
			serviceManager.getLog().warn(RequestMessages.getUnknownConsumerPausingRequestMessage(senderPubKey));
			return;
		}
		
		LocalWorkerProviderClient lwpc = loggedUser.getClient();
		
		DeploymentID lwpcOID = serviceManager.getStubDeploymentID(lwpc);
		if(isRequestUnknown( senderPubKey, request )) {
			serviceManager.getLog().warn(RequestMessages.getUnknownRequestPauseMessage(requestID, lwpcOID));
			return;
		}
		
		if(!request.isPaused()) {
			serviceManager.getLog().debug(RequestMessages.getRequestPausedMessage(requestID, lwpcOID));
			
			Future<?> future = getRequestDAO().removeFuture(request.getSpec());
			if (future != null) {
				future.cancel(true);
			}
			request.pause();
		} else {
			serviceManager.getLog().warn(RequestMessages.getRequestAlreadyPausedMessage(requestID, lwpcOID));
		}
	}

	@Req("REQ118")
	public void resumeRequest(long requestID) {
		
		String senderPubKey = serviceManager.getSenderPublicKey();
		
		if (!UserControl.getInstance().userExists(senderPubKey, getUsersDAO())) {
			serviceManager.getLog().warn(RequestMessages.getUnknownConsumerResumingRequestMessage(senderPubKey));
			return;
		}
		
		Request request = getRequestDAO().getRequest(requestID);
		PeerUserReference loggedUser = getUsersDAO().getLoggedUser(senderPubKey);
		
		if (loggedUser == null) {
			serviceManager.getLog().warn(RequestMessages.getUnknownConsumerResumingRequestMessage(senderPubKey));
			return;
		}
		
		LocalWorkerProviderClient lwpc = loggedUser.getClient();
		
		DeploymentID lwpcOID = serviceManager.getStubDeploymentID(lwpc);
		if(isRequestUnknown( senderPubKey, request )) {
			serviceManager.getLog().warn(RequestMessages.getUnknownRequestResumedMessage(requestID, lwpcOID));
			return;
		}
		
		if(request.isPaused()) {
			serviceManager.getLog().debug(RequestMessages.getRequestResumedMessage(requestID, lwpcOID));
			
			request.resume();
			if(request.needMoreWorkers()) {
				scheduleRequest(request.getSpec());
			}
		} else {
			serviceManager.getLog().warn(RequestMessages.getNotPausedRequestMessage(requestID, lwpcOID));
		}
		
	}

	private RequestDAO getRequestDAO() {
		return serviceManager.getDAO(RequestDAO.class);
	}
	
	private UsersDAO getUsersDAO() {
		return serviceManager.getDAO(UsersDAO.class);
	}

	private ConsumerDAO getConsumerDAO() {
		return serviceManager.getDAO(ConsumerDAO.class);
	}
	
	@Req("REQ014")
	public void finishRequest(RequestSpec requestSpec) {
		
		String brokerPublicKey = serviceManager.getSenderPublicKey();
		
		PeerUser user = UserControl.getInstance().getUserByPublicKey(brokerPublicKey, serviceManager.getDAO(UsersDAO.class));
		
		if (user == null) {
			serviceManager.getLog().warn(RequestMessages.getUnknownUserFinishingRequestMessage(brokerPublicKey));
			return;
		}
		
		RequestDAO requestDAO = getRequestDAO();
		Request request = requestDAO.getRequest(requestSpec.getRequestId());

		PeerUserReference loggedUser = serviceManager.getDAO(UsersDAO.class).getLoggedUser(brokerPublicKey);
		
		if (loggedUser == null) {
			serviceManager.getLog().warn(RequestMessages.getUnknownUserFinishingRequestMessage(brokerPublicKey));
			return;
		}
		
		LocalWorkerProviderClient lwpc = loggedUser.getClient();
		
		if (isRequestUnknown( brokerPublicKey, request )) {
			DeploymentID lwpcOID = serviceManager.getStubDeploymentID(lwpc);
			serviceManager.getLog().warn(RequestMessages.getUnknownRequestFinishedMessage(requestSpec, lwpcOID));
			return;
		}
		
		requestDAO.removeRequest(requestSpec.getRequestId());
		Future<?> future = requestDAO.removeFuture(requestSpec);
		
		List<Request> runningRequests = requestDAO.getRunningRequests();

		boolean hasAnotherRequest = false;
		for (Request runningRequest: runningRequests) {
			if (runningRequest.getConsumer().getPublicKey().equals(brokerPublicKey)) {
				hasAnotherRequest = true;
				break;
			}
		}
		
		if (!hasAnotherRequest) {
			getConsumerDAO().removeLocalConsumer(brokerPublicKey);
		}
		
		if (future != null) {
			future.cancel(true);
		}
		
		List<AllocableWorker> allocableWorkers = new LinkedList<AllocableWorker>(request.getAllocableWorkers());
		
		for (AllocableWorker worker : allocableWorkers) {
			RedistributionController.getInstance().redistributeWorker(worker, serviceManager);
		}
		
		AccountingCommitController.getInstance().commitAccounting(requestSpec, serviceManager);
		
		JobControl.getInstance().finishRequest(request, false, serviceManager);
	}

	@Req("REQ015")
	public void disposeWorker(ServiceID workerServiceID) {
		
		String brokerPublicKey = serviceManager.getSenderPublicKey();
		
		LocalWorkerProviderClient lwpc = validateUser(brokerPublicKey);
		if (lwpc == null) {
			return;
		}
		
		AllocableWorker allocable = validateWorker(workerServiceID, brokerPublicKey, lwpc);
		if (allocable == null) {
			return;
		}
		
		DeploymentID lwpcID = serviceManager.getStubDeploymentID(lwpc);
		
		serviceManager.getLog().debug(WorkerMessages.getDisposingWorkerMessage(allocable.getRequest().getSpec().getRequestId(), 
				lwpcID, workerServiceID));
		
		RedistributionController.getInstance().redistributeWorker(allocable, serviceManager);
	}

	@Req("REQ015")
	private LocalWorkerProviderClient validateUser(String brokerPublicKey) {
		PeerUser user = UserControl.getInstance().getUserByPublicKey(brokerPublicKey, serviceManager.getDAO(UsersDAO.class));
		
		if (user == null) {
			serviceManager.getLog().warn(WorkerMessages.getUnknownConsumerDisposingWorkerMessage(brokerPublicKey));
			return null;
		}
		
		PeerUserReference loggedUser = serviceManager.getDAO(UsersDAO.class).getLoggedUser(brokerPublicKey);
		if (loggedUser == null) {
			serviceManager.getLog().warn(WorkerMessages.getUnknownConsumerDisposingWorkerMessage(brokerPublicKey));
			return null;
		}
		
		LocalWorkerProviderClient lwpc = loggedUser.getClient();
		
		if (lwpc == null) {
			serviceManager.getLog().warn(WorkerMessages.getNotLoggedUserDisposingWorkerMessage(brokerPublicKey));
			return null;
		}
		return lwpc;
	}

	@Req("REQ015")
	private AllocableWorker validateWorker(ServiceID workerServiceID, String myGridPublicKey, LocalWorkerProviderClient lwpc) {
		
		DeploymentID lwpcOID = serviceManager.getStubDeploymentID(lwpc);
		
		if (workerServiceID == null) {
			serviceManager.getLog().warn(WorkerMessages.getNullWorkerDisposalMessage(lwpcOID));
			return null;
		}

		LocalWorker localWorker = WorkerControl.getInstance().getRecoveredWorker(
				workerServiceID, serviceManager.getDAO(LocalWorkersDAO.class));
		
		if (localWorker != null && 
				(localWorker.getStatus().equals(LocalWorkerState.OWNER) || 
						localWorker.getStatus().equals(LocalWorkerState.IDLE))) {
			
//			DeploymentID localWorkerID = serviceManager.getStubDeploymentID(localWorker.getWorkerManagement());
			
			serviceManager.getLog().warn(
					WorkerMessages.getConsumerDisposingNotAllocatedWorkerMessage(
							workerServiceID, lwpcOID));
			return null;
		}
		
		AllocationDAO allocationDAO = serviceManager.getDAO(AllocationDAO.class);
		AllocableWorker allocable = allocationDAO.getAllocableWorker(workerServiceID.getPublicKey());
		
		if (allocable == null) {
			serviceManager.getLog().warn(WorkerMessages.getUnknownWorkerDisposalMessage(lwpcOID));
			return null;
		}
		
//		DeploymentID workerManagementID = serviceManager.getStubDeploymentID(allocable.getWorkerManagement());
		
		if (allocable.getConsumer() == null || allocable.getConsumer().getPublicKey() == null) {
			return null; //TODO
		}
		
		if (!allocable.getConsumer().getPublicKey().equals(myGridPublicKey)) {
			serviceManager.getLog().warn(WorkerMessages.getConsumerDisposingNotAllocatedWorkerMessage(
					workerServiceID, lwpcOID));
			
			return null;
		}
		
		return allocable;
	}

	@Req("REQ016")
	public void unwantedWorker(ServiceID workerID, RequestSpec requestSpec) {
		
		String myGridPublicKey = serviceManager.getSenderPublicKey();
		
		PeerUser user = UserControl.getInstance().getUserByPublicKey(myGridPublicKey, serviceManager.getDAO(UsersDAO.class));
		
		if (user == null) {
			serviceManager.getLog().warn(WorkerMessages.getUnknownConsumerMarkedUnwantedWorkerMessage(myGridPublicKey));
			return;
		}
		
		PeerUserReference loggedUser = serviceManager.getDAO(UsersDAO.class).getLoggedUser(myGridPublicKey);
		
		if (loggedUser == null) {
			serviceManager.getLog().warn(WorkerMessages.getUnknownConsumerMarkedUnwantedWorkerMessage(myGridPublicKey));
			return;
		}
		
		LocalWorkerProviderClient lwpc = loggedUser.getClient();
		DeploymentID lwpcObjectID = serviceManager.getStubDeploymentID(lwpc);
		
		if (workerID == null) {
			serviceManager.getLog().warn(WorkerMessages.getNullUnwantedWorkerMessage(lwpcObjectID));
			return;
		}
		
		AllocableWorker allocable = serviceManager.getDAO(AllocationDAO.class).getAllocableWorker(workerID.getPublicKey());
		
		if (allocable == null) {
			serviceManager.getLog().warn(WorkerMessages.getUnknownUnwantedWorkerMessage(lwpcObjectID, workerID));
			return;
		}
		
		if (!allocable.getConsumer().getDeploymentID().equals(lwpcObjectID)) {
			serviceManager.getLog().warn(WorkerMessages.getNotAllocatedUnwantedWorkerMessage(lwpcObjectID, workerID));
			return;
		}
		
		if (requestSpec == null) {
			serviceManager.getLog().warn(WorkerMessages.getNullRequestUnwantedWorkerMessage(lwpcObjectID, workerID,
					allocable.isWorkerLocal()));
			return;
		}
		
		if (!allocable.getRequest().getSpec().equals(requestSpec)) {
			serviceManager.getLog().warn(WorkerMessages.getInvalidRequestUnwantedWorkerMessage(requestSpec.getRequestId(), lwpcObjectID,
					workerID, allocable.isWorkerLocal()));
			return;
		}
		
		Request request = getRequestDAO().getRequest(requestSpec.getRequestId());
		request.addUnwantedWorker(allocable.getWorkerSpec());
		
		RedistributionController.getInstance().redistributeWorker(allocable, serviceManager);
	}
	
}